package com.yuan;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Arrays;
import java.util.Scanner;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCount {

    public static void main(String[] args) throws Exception {
        // 创建 Hadoop 配置对象
        Configuration conf = new Configuration();

        // 使用 Scanner 从控制台读取用户输入
        Scanner scanner = new Scanner(System.in);

        // 输入文件路径提示与获取
        System.out.print("请输入输入文件路径(如hdfs://localhost:9000/user/hadoop/input)：");
        String inputPaths = scanner.nextLine();

        // 输出文件路径提示与获取
        System.out.print("请输入输出文件路径(如hdfs://localhost:9000/user/hadoop/output）：");
        String outputPath = scanner.nextLine();

        // 创建 MapReduce 作业实例
        Job job = Job.getInstance(conf, "Word Count");

        // 设置作业主类（包含 main 方法的类）
        job.setJarByClass(WordCount.class);

        // 设置 Mapper 类、Combiner 类和 Reducer 类
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(IntSumReducer.class); // Combiner 用于优化性能
        job.setReducerClass(IntSumReducer.class);

        // 设置输出键值对类型
        job.setOutputKeyClass(Text.class); // 输出键为单词（Text 类型）
        job.setOutputValueClass(IntWritable.class); // 输出值为计数（IntWritable 类型）

        // 添加输入路径（支持多个路径用逗号分隔）
        for (String path : inputPaths.split(",")) {
            FileInputFormat.addInputPath(job, new Path(path.trim())); // 去除空格并添加路径
        }

        // 处理输出路径
        Path outputDir = new Path(outputPath);
        try {
            FileSystem fs = outputDir.getFileSystem(conf); // 获取 HDFS 文件系统
            if (fs.exists(outputDir)) { // 检查输出目录是否存在
                fs.delete(outputDir, true); // 如果存在，则递归删除目录
            }
        } catch (IOException e) {
            System.err.println("无法删除输出目录：" + e.getMessage());
            System.exit(1); // 删除失败时退出程序
        }
        FileOutputFormat.setOutputPath(job, outputDir); // 设置输出路径

        // 执行作业并处理结果
        boolean success = job.waitForCompletion(true); // 提交作业并等待完成
        if (success) {
            System.out.println("作业执行成功！");
            int total = countTotalWords(conf, outputDir); // 统计总单词数
            System.out.println("总单词数：" + total); // 在控制台输出统计结果
        } else {
            System.out.println("作业执行失败！");
        }

        scanner.close(); // 关闭 Scanner 对象
    }

    /**
     * 统计输出目录中所有文件的单词总数
     *
     * @param conf      Hadoop 配置对象
     * @param outputDir 输出目录路径
     * @return 总单词数
     * @throws IOException 如果发生 I/O 错误
     */
    private static int countTotalWords(Configuration conf, Path outputDir) throws IOException {
        int total = 0; // 初始化总单词数
        FileSystem fs = outputDir.getFileSystem(conf); // 获取 HDFS 文件系统
        RemoteIterator<LocatedFileStatus> files = fs.listFiles(outputDir, false); // 列出输出目录中的文件

        // 遍历输出目录中的每个文件
        while (files.hasNext()) {
            LocatedFileStatus file = files.next(); // 获取当前文件
            try (FSDataInputStream in = fs.open(file.getPath())) { // 打开文件流
                BufferedReader reader = new BufferedReader(new InputStreamReader(in)); // 创建缓冲读取器
                String line;
                while ((line = reader.readLine()) != null) { // 按行读取文件内容
                    String[] parts = line.split("\t"); // 按制表符分割每行数据
                    System.out.println(Arrays.toString(parts)); // 打印分割后的单词和计数（调试用）
                    if (parts.length == 2) { // 确保分割后有两部分（单词和计数）
                        total += Integer.parseInt(parts[1]); // 将计数累加到总单词数
                    }
                }
            }
        }
        return total; // 返回总单词数
    }

    /**
     * TokenizerMapper 类：实现 Map 阶段的逻辑
     * 将输入文本拆分为单词，并输出 <单词, 1> 的键值对
     */
    public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
        private static final IntWritable ONE = new IntWritable(1); // 常量值 1
        private Text word = new Text(); // 存储单词的变量

        /**
         * Map 方法：将输入文本拆分为单词并输出 <单词, 1>
         *
         * @param key     输入键（通常为偏移量）
         * @param value   输入值（一行文本）
         * @param context 上下文对象，用于写入输出
         * @throws IOException          如果发生 I/O 错误
         * @throws InterruptedException 如果线程被中断
         */
        public void map(Object key, Text value, Context context)
                throws IOException, InterruptedException {
            StringTokenizer tokenizer = new StringTokenizer(value.toString()); // 按空格拆分文本
            while (tokenizer.hasMoreTokens()) { // 遍历每个单词
                word.set(tokenizer.nextToken()); // 获取下一个单词
                context.write(word, ONE); // 输出 <单词, 1>
            }
        }
    }

    /**
     * IntSumReducer 类：实现 Reduce 阶段的逻辑
     * 对相同单词的计数值进行求和，输出 <单词, 总数> 的键值对
     */
    public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable(); // 存储最终计数值的变量

        /**
         * Reduce 方法：对相同单词的计数值进行求和
         *
         * @param key     单词（输入键）
         * @param values  计数值列表（输入值）
         * @param context 上下文对象，用于写入输出
         * @throws IOException          如果发生 I/O 错误
         * @throws InterruptedException 如果线程被中断
         */
        public void reduce(Text key, Iterable<IntWritable> values, Context context)
                throws IOException, InterruptedException {
            int sum = 0; // 初始化计数总和
            for (IntWritable val : values) { // 遍历计数值列表
                sum += val.get(); // 累加计数值
            }
            result.set(sum); // 设置最终计数结果
            context.write(key, result); // 输出 <单词, 总数>
        }
    }
}